package com.htdata.iiot;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.alibaba.fastjson.JSONObject;
import com.htdata.iiot.api.pojo.DeviceNumber;
import com.htdata.iiot.api.pojo.Metric;
import com.htdata.iiot.api.pojo.StatusData;
import com.htdata.iiot.api.pojo.StreamingData;
import com.htdata.iiot.api.pojo.StreamingMetricData;
import com.htdata.iiot.db.DbConnectionPool;
import com.htdata.iiot.db.JDBCUtils;
import com.htdata.iiot.utils.FileUtils;

/**
 * 
 * 
 */
public class Main15 {
	private static Logger logger = LogManager.getLogger(Main15.class);
	private static DbConnectionPool pool;
	private static String tenantIds;
	private static int times=60;
	private static Map<String, Metric> dnTometric = new HashMap<>();

	public static void main(String[] args) throws ParseException {
		Cmd.parserCmd(args);
		tenantIds = Cmd.tenantIds;
		times = Cmd.times;
		
		pool  = DbConnectionPool.getPool();
		List<StatusData> sList = new ArrayList<StatusData>();
		SendDataToKafka sendDataToKafka = new SendDataToKafka();
		// 1.根据租户ID获取设备

		String sql1 = " select * from (SELECT  T2.* FROM" + "  (SELECT" + "    T.*," + "    CASE" + "      WHEN @mid = struct_id"
				+ "      THEN @row := @row + 1" + "      ELSE @row := 1" + "    END rownum,"
				+ "    @mid := struct_id MID" + "  FROM" + "    (SELECT" + "      struct_id," + "      id,tenant_id,"
				+ "      SUM(1)" + "    FROM" + "      thing_device" + "      WHERE"
				+ "        deleted=0 AND TYPE=0" + "    GROUP BY struct_id,id,tenant_id"
				+ "    ORDER BY struct_id,id," + "      SUM(1) DESC) AS T) AS T2 ) t " ;
		if(tenantIds!=null&&!tenantIds.trim().isEmpty()){
			sql1+=" where `tenant_id` in("+tenantIds+") ";
		}
		
		if(Cmd.pageSize>0){
			sql1 += "ORDER BY rownum,id " + "LIMIT 0,"+Cmd.pageSize+" ";
		}


		logger.info(sql1);
		Connection conn = pool.getConnection();
		PreparedStatement ps = null;
		ResultSet rs = null;
		ResultSet executeQuery = null;
		PreparedStatement prepareStatement = null;
		try {
			ps = conn.prepareStatement(sql1);
			rs = ps.executeQuery();
			while (rs.next()) {
				int deviceId = rs.getInt("id");
				int tenantId = rs.getInt("tenant_id");

				StatusData statusData = new StatusData();
				statusData.setTid(tenantId);
				statusData.setDid(deviceId);
				sList.add(statusData);
			}
		} catch (SQLException e) {
			e.printStackTrace();
		} finally {
			JDBCUtils.close(ps, rs, conn);
			JDBCUtils.close(prepareStatement, executeQuery, null);
		}

		logger.info("****************************");
		logger.info(sList.size());
		logger.info("****************************");
		long startTime = System.currentTimeMillis() - 1;
		long st = System.currentTimeMillis();
		long i = 0;
		long mi = 0;
		Object value = null;
		long avg = sList.size();
		if(times!=0){
			avg = sList.size()/(times*2);
		}
		if(avg == 0){
			avg = sList.size();
		}
		
		while (true) {
			
			// 3.生成数据
			for (StatusData sData : sList) {
				
				long currentTimeMillis = System.currentTimeMillis();
				sData.setTs(currentTimeMillis);
				long m = currentTimeMillis%60;
				long status = 1000;
				float nextFloat = new Random().nextFloat();
				float f = 50+10*nextFloat;
				if(m<=f){
					status=1000;
				}else if(m<=f+5){
					status=2000;
				}else if(m<=f+8){
					status=3000;
				}else{
					status=3000;
				}
				
				
				sData.setSts(status);
				System.out.println(buildResult(sData));
				sendDataToKafka.sendStat(sData.getDid(),buildResult(sData));
				i++;
				if(i%avg==0){
					//线程休息
					try {
						logger.info("中场休息"+"i:"+i);
						Thread.sleep(400);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
			}
			
			long endTime = System.currentTimeMillis();
			long useTime = endTime - startTime;
			logger.info(st+"写入 " + i + " 条数据,累计测点数：" + mi + "，平均速率 " + (i * 1000 / useTime) + "，测点速率 "
					+ (mi * 1000 / useTime));
			if(endTime<(st+times*1000)){
				try {
					logger.info("本场结束");
					Thread.sleep(st+times*1000-endTime);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}else{
				logger.info("本场超时");
			}

		}

	}

	private static String buildResult(StatusData sData) {
		return JSONObject.toJSONString(sData).toString();
	}

}